--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit a9cc4b996b8da8ff414e305825f372ca12dc138d
Parents : bce42bc
Author : Ivan <e46112d44649266d71fe2193e00a4710>
Signature : T66BB85Valid, signed by author
Date : 2026-07-19T08:55:50-05:00
feat: fix auto-resend functionality with attachment handling and improve duplicate message cleanup logic
Changes
7 files changed, 425 insertions(+), 36 deletions(-)
Diff
diff --git a/meshchatx.rsm b/meshchatx.rsm
index 259c26d4..48ea0950 100644
Binary files a/meshchatx.rsm and b/meshchatx.rsm differ
diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py
index e7e50242..019ec8f8 100644
--- a/meshchatx/meshchat.py
+++ b/meshchatx/meshchat.py
@@ -183,8 +183,10 @@ from meshchatx.src.backend.auto_resend_guard import (
RECENT_SAME_CONTENT_SECONDS,
AutoResendCoordinator,
cooldown_until,
+ fields_have_attachments,
fields_with_auto_resend_count,
next_attempt_count,
+ parse_fields_dict,
should_skip_for_budget,
)
from meshchatx.src.backend.local_message_retention import (
@@ -23244,6 +23246,16 @@ class ReticulumMeshChat:
):
continue
+ fields = parse_fields_dict(failed_message.get("fields"))
+ allow_attachments = (
+ ctx.config.allow_auto_resending_failed_messages_with_attachments.get()
+ )
+ if not allow_attachments and fields_have_attachments(fields):
+ print(
+ "Not resending failed message with attachments, as setting is disabled",
+ )
+ continue
+
claimed = (
ctx.database.messages.try_claim_failed_message_for_auto_resend(
message_hash,
@@ -23257,14 +23269,9 @@ class ReticulumMeshChat:
if not claimed:
continue
- # parse fields as json
- fields = json.loads(failed_message["fields"] or "{}")
- if not isinstance(fields, dict):
- fields = {}
-
# parse image field
image_field = None
- if "image" in fields:
+ if "image" in fields and isinstance(fields.get("image"), dict):
image_field = LxmfImageField(
fields["image"]["image_type"],
base64.b64decode(fields["image"]["image_bytes"]),
@@ -23272,7 +23279,7 @@ class ReticulumMeshChat:
# parse audio field
audio_field = None
- if "audio" in fields:
+ if "audio" in fields and isinstance(fields.get("audio"), dict):
audio_field = LxmfAudioField(
fields["audio"]["audio_mode"],
base64.b64decode(fields["audio"]["audio_bytes"]),
@@ -23280,30 +23287,22 @@ class ReticulumMeshChat:
# parse file attachments field
file_attachments_field = None
- if "file_attachments" in fields:
+ if "file_attachments" in fields and isinstance(
+ fields.get("file_attachments"),
+ list,
+ ):
file_attachments = [
LxmfFileAttachment(
file_attachment["file_name"],
base64.b64decode(file_attachment["file_bytes"]),
)
for file_attachment in fields["file_attachments"]
+ if isinstance(file_attachment, dict)
]
file_attachments_field = LxmfFileAttachmentsField(
file_attachments,
)
- # don't resend message with attachments if not allowed
- if not ctx.config.allow_auto_resending_failed_messages_with_attachments.get():
- if (
- image_field is not None
- or audio_field is not None
- or file_attachments_field is not None
- ):
- print(
- "Not resending failed message with attachments, as setting is disabled",
- )
- continue
-
attempt = next_attempt_count(failed_message.get("fields"))
ctx.database.messages.set_message_fields_json(
message_hash,
diff --git a/meshchatx/src/backend/auto_resend_guard.py b/meshchatx/src/backend/auto_resend_guard.py
index 611035d3..9ce2e91e 100644
--- a/meshchatx/src/backend/auto_resend_guard.py
+++ b/meshchatx/src/backend/auto_resend_guard.py
@@ -67,6 +67,20 @@ class AutoResendCoordinator:
return lock
+def fields_have_attachments(fields_raw: Any) -> bool:
+ fields = _parse_fields(fields_raw)
+ if not fields:
+ return False
+ if fields.get("image") or fields.get("audio"):
+ return True
+ files = fields.get("file_attachments")
+ return isinstance(files, list) and len(files) > 0
+
+
+def parse_fields_dict(fields_raw: Any) -> dict:
+ return _parse_fields(fields_raw)
+
+
def should_skip_for_budget(
fields_raw: Any, *, max_attempts: int = MAX_AUTO_RESEND_ATTEMPTS
) -> bool:
diff --git a/meshchatx/src/backend/database/messages.py b/meshchatx/src/backend/database/messages.py
index 76185a7e..4f650826 100644
--- a/meshchatx/src/backend/database/messages.py
+++ b/meshchatx/src/backend/database/messages.py
@@ -852,26 +852,38 @@ class MessageDAO:
"""Hashes of duplicate rows (same peer, direction, and text), excluding the oldest keep.
Empty or whitespace-only content is ignored so attachment-only or blank
- rows are not collapsed together.
+ rows are not collapsed together. Oldest is by timestamp then id.
"""
- rows = self.provider.fetchall(
+ groups = self.provider.fetchall(
"""
- SELECT m.hash AS hash
- FROM lxmf_messages m
- INNER JOIN (
- SELECT peer_hash, is_incoming, content, MIN(id) AS keep_id
- FROM lxmf_messages
- WHERE content IS NOT NULL AND TRIM(content) != ''
- GROUP BY peer_hash, is_incoming, content
- HAVING COUNT(*) > 1
- ) d
- ON m.peer_hash = d.peer_hash
- AND m.is_incoming = d.is_incoming
- AND m.content = d.content
- WHERE m.id != d.keep_id
+ SELECT peer_hash, is_incoming, content
+ FROM lxmf_messages
+ WHERE content IS NOT NULL AND TRIM(content) != ''
+ GROUP BY peer_hash, is_incoming, content
+ HAVING COUNT(*) > 1
""",
)
- return [r["hash"] for r in rows if r.get("hash")]
+ to_delete: list[str] = []
+ for group in groups:
+ rows = self.provider.fetchall(
+ """
+ SELECT hash
+ FROM lxmf_messages
+ WHERE peer_hash = ?
+ AND is_incoming = ?
+ AND content = ?
+ ORDER BY
+ CASE WHEN timestamp IS NULL THEN 1 ELSE 0 END,
+ timestamp ASC,
+ id ASC
+ """,
+ (group["peer_hash"], group["is_incoming"], group["content"]),
+ )
+ # Keep the first (oldest); delete the rest.
+ for row in rows[1:]:
+ if row.get("hash"):
+ to_delete.append(row["hash"])
+ return to_delete
def count_duplicate_lxmf_messages_by_content(self) -> int:
return len(self.list_duplicate_lxmf_message_hashes_by_content())
diff --git a/tests/backend/test_auto_resend_regression.py b/tests/backend/test_auto_resend_regression.py
new file mode 100644
index 00000000..b81bfb50
--- /dev/null
+++ b/tests/backend/test_auto_resend_regression.py
@@ -0,0 +1,338 @@
+# SPDX-License-Identifier: 0BSD
+
+"""Regression tests for auto-resend flooding guards and duplicate cleanup."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+import types
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from meshchatx.meshchat import ReticulumMeshChat
+from meshchatx.src.backend import auto_resend_guard as guard
+from meshchatx.src.backend.database import Database
+from meshchatx.src.backend.database.provider import DatabaseProvider
+from meshchatx.src.backend.database.schema import DatabaseSchema
+
+
+@pytest.fixture
+def db(tmp_path):
+ path = str(tmp_path / "resend_reg.db")
+ provider = DatabaseProvider(path)
+ DatabaseSchema(provider).initialize()
+ database = Database(path)
+ yield database
+ database.close_all()
+ provider.close_all()
+
+
+def _insert_failed(db, *, msg_hash, peer, content, fields="{}", ts=None, next_at=None):
+ now = ts if ts is not None else time.time()
+ db.messages.upsert_lxmf_message(
+ {
+ "hash": msg_hash,
+ "source_hash": peer,
+ "destination_hash": peer,
+ "peer_hash": peer,
+ "state": "failed",
+ "progress": 0,
+ "is_incoming": 0,
+ "method": "opportunistic",
+ "delivery_attempts": 0,
+ "next_delivery_attempt_at": next_at,
+ "title": "",
+ "content": content,
+ "fields": fields,
+ "rssi": None,
+ "snr": None,
+ "quality": None,
+ "is_spam": 0,
+ "reply_to_hash": None,
+ "attachments_stripped": 0,
+ "timestamp": now,
+ },
+ )
+
+
+def _insert_outbound(db, *, msg_hash, peer, content, state="delivered", ts=None):
+ now = ts if ts is not None else time.time()
+ db.messages.upsert_lxmf_message(
+ {
+ "hash": msg_hash,
+ "source_hash": peer,
+ "destination_hash": peer,
+ "peer_hash": peer,
+ "state": state,
+ "progress": 1,
+ "is_incoming": 0,
+ "method": "direct",
+ "delivery_attempts": 1,
+ "next_delivery_attempt_at": None,
+ "title": "",
+ "content": content,
+ "fields": "{}",
+ "rssi": None,
+ "snr": None,
+ "quality": None,
+ "is_spam": 0,
+ "reply_to_hash": None,
+ "attachments_stripped": 0,
+ "timestamp": now,
+ },
+ )
+
+
+def _bind_resend_app(db):
+ app = MagicMock()
+ app._auto_resend_coordinator = guard.AutoResendCoordinator()
+ app.websocket_broadcast = AsyncMock()
+ app.send_message = AsyncMock()
+ app.resend_failed_messages_for_destination = types.MethodType(
+ ReticulumMeshChat.resend_failed_messages_for_destination,
+ app,
+ )
+ ctx = MagicMock()
+ ctx.identity.hash.hex.return_value = "aa" * 16
+ ctx.database = db
+ ctx.config.allow_auto_resending_failed_messages_with_attachments.get.return_value = (
+ True
+ )
+ return app, ctx
+
+
+def test_fields_have_attachments_and_invalid_json():
+ assert not guard.fields_have_attachments("{}")
+ assert not guard.fields_have_attachments("not-json")
+ assert guard.fields_have_attachments('{"image":{"image_type":"png"}}')
+ assert guard.fields_have_attachments('{"file_attachments":[{"file_name":"a"}]}')
+ assert guard.parse_fields_dict("not-json") == {}
+ assert guard.should_skip_for_budget(
+ '{"_mcx_auto_resend_count": 3}',
+ max_attempts=3,
+ )
+
+
+@pytest.mark.asyncio
+async def test_resend_skips_when_budget_exhausted(db):
+ peer = "b" * 32
+ msg = "c" * 32
+ _insert_failed(
+ db,
+ msg_hash=msg,
+ peer=peer,
+ content="help",
+ fields=json.dumps({"_mcx_auto_resend_count": 3}),
+ )
+ app, ctx = _bind_resend_app(db)
+ await app.resend_failed_messages_for_destination(peer, context=ctx)
+ app.send_message.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_resend_skips_when_recent_same_content_exists(db):
+ peer = "d" * 32
+ now = time.time()
+ _insert_outbound(
+ db,
+ msg_hash="1" * 32,
+ peer=peer,
+ content="help text",
+ ts=now - 30,
+ )
+ _insert_failed(
+ db,
+ msg_hash="2" * 32,
+ peer=peer,
+ content="help text",
+ ts=now - 5,
+ )
+ app, ctx = _bind_resend_app(db)
+ await app.resend_failed_messages_for_destination(peer, context=ctx)
+ app.send_message.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_resend_skips_attachments_before_claim(db):
+ peer = "e" * 32
+ msg = "f" * 32
+ fields = json.dumps(
+ {
+ "image": {
+ "image_type": "png",
+ "image_bytes": "aa",
+ },
+ },
+ )
+ _insert_failed(db, msg_hash=msg, peer=peer, content="pic", fields=fields)
+ app, ctx = _bind_resend_app(db)
+ ctx.config.allow_auto_resending_failed_messages_with_attachments.get.return_value = (
+ False
+ )
+ await app.resend_failed_messages_for_destination(peer, context=ctx)
+ app.send_message.assert_not_called()
+ row = db.provider.fetchone(
+ "SELECT next_delivery_attempt_at FROM lxmf_messages WHERE hash = ?",
+ (msg,),
+ )
+ # Must not burn cooldown when attachments are disallowed.
+ assert row["next_delivery_attempt_at"] is None
+
+
+@pytest.mark.asyncio
+async def test_resend_claims_once_and_deletes_old_on_success(db):
+ peer = "1" * 32
+ old_hash = "2" * 32
+ new_hash_bytes = bytes.fromhex("3" * 32)
+ _insert_failed(db, msg_hash=old_hash, peer=peer, content="retry me")
+ app, ctx = _bind_resend_app(db)
+
+ async def send_and_materialize(*args, **kwargs):
+ new_msg = MagicMock()
+ new_msg.hash = new_hash_bytes
+ _insert_outbound(
+ db,
+ msg_hash=new_hash_bytes.hex(),
+ peer=peer,
+ content="retry me",
+ state="outbound",
+ )
+ return new_msg
+
+ app.send_message.side_effect = send_and_materialize
+
+ await app.resend_failed_messages_for_destination(peer, context=ctx)
+ assert app.send_message.await_count == 1
+ assert (
+ db.provider.fetchone(
+ "SELECT 1 FROM lxmf_messages WHERE hash = ?",
+ (old_hash,),
+ )
+ is None
+ )
+ new_row = db.provider.fetchone(
+ "SELECT fields FROM lxmf_messages WHERE hash = ?",
+ (new_hash_bytes.hex(),),
+ )
+ assert guard.read_auto_resend_count(new_row["fields"]) == 1
+ app.websocket_broadcast.assert_awaited()
+
+
+@pytest.mark.asyncio
+async def test_resend_concurrent_calls_send_only_once(db):
+ peer = "4" * 32
+ old_hash = "5" * 32
+ _insert_failed(db, msg_hash=old_hash, peer=peer, content="race")
+ app, ctx = _bind_resend_app(db)
+
+ async def slow_send(*args, **kwargs):
+ await asyncio.sleep(0.05)
+ m = MagicMock()
+ m.hash = bytes.fromhex("6" * 32)
+ _insert_outbound(
+ db,
+ msg_hash=m.hash.hex(),
+ peer=peer,
+ content="race",
+ state="outbound",
+ )
+ return m
+
+ app.send_message.side_effect = slow_send
+ await asyncio.gather(
+ app.resend_failed_messages_for_destination(peer, context=ctx),
+ app.resend_failed_messages_for_destination(peer, context=ctx),
+ )
+ assert app.send_message.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_resend_does_not_delete_old_when_send_returns_none(db):
+ peer = "7" * 32
+ old_hash = "8" * 32
+ _insert_failed(db, msg_hash=old_hash, peer=peer, content="keep")
+ app, ctx = _bind_resend_app(db)
+ app.send_message.return_value = None
+ await app.resend_failed_messages_for_destination(peer, context=ctx)
+ assert db.provider.fetchone(
+ "SELECT state FROM lxmf_messages WHERE hash = ?",
+ (old_hash,),
+ )["state"] == "failed"
+ # Cooldown claimed, attempt counted.
+ row = db.provider.fetchone(
+ "SELECT fields, next_delivery_attempt_at FROM lxmf_messages WHERE hash = ?",
+ (old_hash,),
+ )
+ assert guard.read_auto_resend_count(row["fields"]) == 1
+ assert row["next_delivery_attempt_at"] is not None
+
+
+def test_duplicate_cleanup_keeps_oldest_by_timestamp(db):
+ peer = "9" * 32
+ base = time.time()
+ # Later insert id can still be oldest by timestamp.
+ _insert_outbound(db, msg_hash="a" * 32, peer=peer, content="dup", ts=base - 5)
+ _insert_outbound(db, msg_hash="b" * 32, peer=peer, content="dup", ts=base - 50)
+ _insert_outbound(db, msg_hash="c" * 32, peer=peer, content="dup", ts=base - 1)
+ deleted = db.messages.delete_duplicate_lxmf_messages_by_content()
+ assert deleted == 2
+ left = db.provider.fetchone(
+ "SELECT hash FROM lxmf_messages WHERE content = 'dup'",
+ )
+ assert left["hash"] == "b" * 32
+
+
+def test_source_keeps_auto_resend_and_duplicate_guards():
+ from pathlib import Path
+
+ root = Path(__file__).resolve().parents[2]
+ meshchat = (root / "meshchatx" / "meshchat.py").read_text(encoding="utf-8")
+ guard_src = (
+ root / "meshchatx" / "src" / "backend" / "auto_resend_guard.py"
+ ).read_text(encoding="utf-8")
+ messages = (
+ root / "meshchatx" / "src" / "backend" / "database" / "messages.py"
+ ).read_text(encoding="utf-8")
+ settings = (
+ root
+ / "meshchatx"
+ / "src"
+ / "frontend"
+ / "components"
+ / "settings"
+ / "SettingsPage.vue"
+ ).read_text(encoding="utf-8")
+
+ assert "AutoResendCoordinator" in meshchat
+ assert "try_claim_failed_message_for_auto_resend" in meshchat
+ assert "fields_have_attachments" in meshchat
+ assert "has_path(destination_hash)" in meshchat
+ assert "MAX_AUTO_RESEND_ATTEMPTS" in guard_src
+ assert "delete_duplicate_lxmf_messages_by_content" in messages
+ assert "clearDuplicateMessages" in settings
+ assert "/api/v1/maintenance/messages/duplicates" in meshchat
+
+
+def test_claim_reopens_after_cooldown_expires(db):
+ peer = "0" * 32
+ msg = "a" * 32
+ now = 1_000_000.0
+ _insert_failed(db, msg_hash=msg, peer=peer, content="x", next_at=now - 1)
+ assert db.messages.try_claim_failed_message_for_auto_resend(
+ msg,
+ cooldown_until=now + 120,
+ now=now,
+ )
+ assert not db.messages.try_claim_failed_message_for_auto_resend(
+ msg,
+ cooldown_until=now + 240,
+ now=now + 10,
+ )
+ assert db.messages.try_claim_failed_message_for_auto_resend(
+ msg,
+ cooldown_until=now + 360,
+ now=now + 121,
+ )
diff --git a/tests/backend/test_duplicate_messages_cleanup.py b/tests/backend/test_duplicate_messages_cleanup.py
index 708c502c..9c1800d8 100644
--- a/tests/backend/test_duplicate_messages_cleanup.py
+++ b/tests/backend/test_duplicate_messages_cleanup.py
@@ -65,6 +65,24 @@ def test_delete_duplicate_messages_by_content_keeps_oldest(tmp_path):
provider.close_all()
+def test_delete_duplicate_prefers_earlier_timestamp_over_lower_id(tmp_path):
+ path = str(tmp_path / "d3.db")
+ provider = DatabaseProvider(path)
+ DatabaseSchema(provider).initialize()
+ db = Database(path)
+ peer = "d" * 32
+ base = time.time()
+ # Insert newer timestamp first (lower id), then older timestamp.
+ _msg(db, msg_hash="1" * 32, peer=peer, content="z", ts=base - 1)
+ _msg(db, msg_hash="2" * 32, peer=peer, content="z", ts=base - 100)
+ deleted = db.messages.delete_duplicate_lxmf_messages_by_content()
+ assert deleted == 1
+ left = db.provider.fetchone("SELECT hash FROM lxmf_messages")
+ assert left["hash"] == "2" * 32
+ db.close_all()
+ provider.close_all()
+
+
def test_duplicates_are_scoped_by_peer_and_direction(tmp_path):
path = str(tmp_path / "d2.db")
provider = DatabaseProvider(path)
diff --git a/tests/frontend/SettingsPage.config-persistence.test.js b/tests/frontend/SettingsPage.config-persistence.test.js
index 4d9993bd..a3f4e14e 100644
--- a/tests/frontend/SettingsPage.config-persistence.test.js
+++ b/tests/frontend/SettingsPage.config-persistence.test.js
@@ -611,6 +611,14 @@ describe("SettingsPage — maintenance, exports, telemetry trust, RNS reload", (
expect(api.delete).toHaveBeenCalledWith("/api/v1/maintenance/messages/duplicates");
});
+ it("maintenance UI exposes duplicate cleanup and age purge", async () => {
+ const w = await mountSettingsPage(api);
+ const html = w.html();
+ expect(html).toContain("maintenance.clear_duplicates");
+ expect(html).toContain("maintenance.purge_old_title");
+ expect(html).toContain("maintenance.export_old_archive");
+ });
+
it("clearAnnounces DELETEs announces", async () => {
const w = await mountSettingsPage(api);
await w.vm.clearAnnounces();
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────